package org.apache.rocketmq.example.transaction;

import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;

import java.nio.charset.StandardCharsets;
import java.util.Arrays;
import java.util.concurrent.*;

public class MyTransactionProducer {
    public static final String TOPIC = "TopicTest1234";
    public static void main(String[] args) throws MQClientException, InterruptedException {
        TransactionMQProducer producer = new TransactionMQProducer("TransactionProducer", Arrays.asList(TOPIC));
        ThreadPoolExecutor executor = new ThreadPoolExecutor(2, 5, 100, TimeUnit.SECONDS, new ArrayBlockingQueue<>(1000),
                new ThreadFactory() {
                    @Override
                    public Thread newThread(Runnable r) {
                        Thread thread = new Thread(r);
                        return thread;
                    }
                });
        producer.setExecutorService(executor);
        producer.setNamesrvAddr("127.0.0.1:9876");
        producer.setTransactionListener(new MyTransaction());
        producer.start();

        Message message = new Message(TOPIC, "Tag1", "hello transcation".getBytes(StandardCharsets.UTF_8));

        producer.sendMessageInTransaction(message,"my args");

        TimeUnit.SECONDS.sleep(10);

        producer.shutdown();

    }
}


class MyTransaction implements TransactionListener {
    @Override
    public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
        System.out.println(msg);
        return LocalTransactionState.COMMIT_MESSAGE;
    }

    @Override
    public LocalTransactionState checkLocalTransaction(MessageExt msg) {
        return LocalTransactionState.COMMIT_MESSAGE;
    }
}